package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String topicName = "topic_log";
        String groupId = "base_log_app";
        DataStreamSource<String> logStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));
        //转换数据结构 json
        SingleOutputStreamOperator<JSONObject> jsonStream = logStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                out.collect(jsonObject);
            }
        });
        //新旧访客修复
        KeyedStream<JSONObject, String> keyedStream = jsonStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("mid");
            }
        });
        SingleOutputStreamOperator<JSONObject> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> isNewFirstFlagState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                isNewFirstFlagState = getRuntimeContext().getState(new ValueStateDescriptor<String>("is_new_first_flag", String.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String isNewFirstFlag = isNewFirstFlagState.value();
                Long ts = value.getLong("ts");
                String toDate = DateFormatUtil.toDate(ts);
                String isNew = value.getJSONObject("common").getString("is_new");
                if ("0".equals(isNew)) {//判断是否旧访客
                    if (isNewFirstFlag == null) {//旧访客状态为空，将前一天的date更新到状态
                        isNewFirstFlagState.update(DateFormatUtil.toDate(ts - 60 * 60 * 24 * 1000L));
                    }
                } else {
                    if (isNewFirstFlag == null) {//新访客状态为空,更新当天日期到状态
                        isNewFirstFlagState.update(toDate);
                    } else if (!isNewFirstFlag.equals(toDate)) {//"新访客"的状态与当天的日期不等，则该访客不是新访客，需要将is_new修改为0
                        value.getJSONObject("common").put("is_new", "0");
                    }
                }
                out.collect(value);
            }
        });
        OutputTag<String> actionOutPutTag = new OutputTag<String>("action"){};
        OutputTag<String> displayOutPutTag = new OutputTag<String>("display"){};
        OutputTag<String> startOutPutTag = new OutputTag<String>("start"){};
        OutputTag<String> errorOutPutTag = new OutputTag<String>("error"){};
        OutputTag<String> appVideoOutPutTag = new OutputTag<String>("appVideo"){};//播放日志

        SingleOutputStreamOperator<String> processedStream = (SingleOutputStreamOperator<String>) processStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                JSONObject err = value.getJSONObject("err");
                Long ts = value.getLong("ts");
                if (err != null) {
                    ctx.output(errorOutPutTag, err.toJSONString());
                    value.remove("err");
                }
                JSONObject start = value.getJSONObject("start");
                JSONObject appVideo = value.getJSONObject("appVideo");
                if (start != null) {
                    ctx.output(startOutPutTag, value.toJSONString());
                } else if (appVideo != null) {
                    ctx.output(appVideoOutPutTag, value.toJSONString());
                } else {
                    JSONObject common = value.getJSONObject("common");
                    JSONArray displays = value.getJSONArray("displays");
                    JSONObject page = value.getJSONObject("page");
                    //曝光日志
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("common", common);
                            display.put("ts", ts);
                            display.put("page", page);
                            ctx.output(displayOutPutTag, display.toJSONString());
                        }
                        value.remove("displays");
                    }
                    JSONArray actions = value.getJSONArray("actions");
                    //动作浏览日志
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("common", common);
                            action.put("page", page);
                            ctx.output(actionOutPutTag, action.toJSONString());
                        }
                        value.remove("actions");
                    }
                    out.collect(value.toJSONString());
                }
            }
        });
        //获取侧输出流
        DataStream<String> actionStream = processedStream.getSideOutput(actionOutPutTag);
        DataStream<String> displayStream = processedStream.getSideOutput(displayOutPutTag);
        DataStream<String> startStream = processedStream.getSideOutput(startOutPutTag);
        DataStream<String> errorStream = processedStream.getSideOutput(errorOutPutTag);
        DataStream<String> appVideoStream = processedStream.getSideOutput(appVideoOutPutTag);

        //测试输出数据流
        /*startStream.print("start>>");
        errorStream.print("error>>");
        displayStream.print("display>>");
        actionStream.print("action>>");
        appVideoStream.print("appVideo>>>");*/
        processedStream.print("page>>>");


        //写出不同kafka主题
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String appVideo_topic = "dwd_traffic_appVideo_log";

        processedStream.addSink(KafkaUtil.getFlinkKafkaProducer(page_topic));
        startStream.addSink(KafkaUtil.getFlinkKafkaProducer(start_topic));
        displayStream.addSink(KafkaUtil.getFlinkKafkaProducer(display_topic));
        actionStream.addSink(KafkaUtil.getFlinkKafkaProducer(action_topic));
        errorStream.addSink(KafkaUtil.getFlinkKafkaProducer(error_topic));
        appVideoStream.addSink(KafkaUtil.getFlinkKafkaProducer(appVideo_topic));

        env.execute(groupId);
    }
}
